package com.atguigu.gmall.realtime.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.util.HbaseUtil;
import com.atguigu.gmall.realtime.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

/**
 * @Author lzc
 * @Date 2023/7/31 10:17
 */
public class HbaseSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection hbaseConn;
    private Jedis redisClient;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseConn = HbaseUtil.getHbaseConnection();
        
        redisClient = RedisUtil.getRedisClient();
    }
    
    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbaseConnection(hbaseConn);
    
        RedisUtil.closeRedisClient(redisClient);
        
    }
    
    // 每来一条数据, 触发一次
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> t,
                       Context context) throws Exception {
        JSONObject data = t.f0;
        String opType = data.getString("op_type");
        // insert update delete
        // 如果是 insert 或 update的时候, 向 hbase 中写一行
        
        // 如果是 delete, 则是删除 hbase 中删除一行
        if ("delete".equals(opType)) {
            deleteOneRow(t);
        } else if ("insert".equals(opType) || "update".equals(opType)) {
            putOneRow(t);
        }
        
        // 缓存删除
        // 当维度删除或者维度更新的时候,都去 redis 中删除缓存
        TableProcess tp = t.f1;
        if ("delete".equals(opType) || "update".equals(opType)) {
            System.out.println("维度发生变化, 删除维度: " + tp.getSinkTable() + "  " + data.getString("id"));
            RedisUtil.deleteOneRowCache(redisClient, tp.getSinkTable(), data.getString("id"));  // key: 表名+id
        }
        
        
    }
    
    private void putOneRow(Tuple2<JSONObject, TableProcess> t) {
        JSONObject data = t.f0;
        TableProcess tp = t.f1;
        HbaseUtil.putOneRow(hbaseConn,
                            "gmall",
                            tp.getSinkTable(),
                            data.getString(tp.getSinkRowKey()),
                            tp.getSinkFamily(),
                            tp.getSinkColumns().split(","),
                            data
        );
    }
    
    private void deleteOneRow(Tuple2<JSONObject, TableProcess> t) {
        JSONObject data = t.f0;
        TableProcess tp = t.f1;
        HbaseUtil.deleteOneRow(hbaseConn,
                               "gmall",
                               tp.getSinkTable(),
                               data.getString(tp.getSinkRowKey())
        );
    }
}
